package marsdonne;

import marsdonne.flink.connectors.RMQSink;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

public class TestDataGenerator {

    static final String testData = "{\"type\":\"resident_manage_status\",\"timed\":\"2018-12-16\",\"metadata\":{\"ver\":1.1},\"contents\":{\"status\":1}}" +
            "|{\"type\":\"resident_manage_status\",\"timed\":\"2018-12-18\",\"metadata\":{\"ver\":1.1},\"contents\":{\"status\":0}}" +
            "|{\"type\":\"resident_manage_status\",\"timed\":\"2018-12-19\",\"metadata\":{\"ver\":1.1},\"contents\":{\"status\":1}}";

    public static void main(String[] args) throws Exception {
        final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
                .setUserName("guest")
                .setPassword("guest")
                .setVirtualHost("/")//multi-tenant
                .setHost("192.168.0.125")
                .setPort(5672)
                .build();

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // checkpointing is required for exactly-once or at-least-once guarantees
        env.enableCheckpointing(1000);//60 * 1000

        env.fromElements(testData.split("\\|"))
                .addSink(new RMQSink<>(
                        connectionConfig,                          // config for the RabbitMQ connection
                        "data.business.increment.forwarding",               // name of the RabbitMQ queue to send messages to
                        new SimpleStringSchema())
                );                                                // serialization schema to turn Java objects to messages

        env.execute("execute generator");
    }
}
